/*-------------------------------------------------------------------------
 *
 * connection.c
 *		  postgres_fdw 的连接管理函数
 *
 * Portions Copyright (c) 2012-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *		  contrib/postgres_fdw/connection.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/htup_details.h"
#include "access/xact.h"
#include "catalog/pg_user_mapping.h"
#include "commands/defrem.h"
#include "funcapi.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postgres_fdw.h"
#include "storage/fd.h"
#include "storage/latch.h"
#include "utils/builtins.h"
#include "utils/datetime.h"
#include "utils/hsearch.h"
#include "utils/inval.h"
#include "utils/memutils.h"
#include "utils/syscache.h"

/*
 * 连接缓存哈希表条目
 *
 * 该哈希表中的查找键是用户映射OID。我们每个用户映射ID只使用一个连接，这确保了在查询期间所有扫描使用相同的快照。使用用户映射OID而不是外部服务器OID + 用户OID，可以避免在公共用户映射适用于所有用户OID时创建多个连接。
 *
 * 如果当前没有活动连接，“conn”指针可以为NULL。当我们有连接时，xact_depth跟踪远程端打开的事务和子事务的当前深度。我们需要在远程执行与我们自己执行相同嵌套深度的命令，以便回滚子事务将终止正确的查询，而不是错误的查询。
 */
typedef Oid ConnCacheKey;

typedef struct ConnCacheEntry
{
	ConnCacheKey key;			/* 哈希键（必须是第一个） */
	PGconn	   *conn;			/* 连接到外部服务器，或NULL */
	/* 当conn为NULL时，剩余字段无效： */
	int			xact_depth;		/* 0 = 无打开的xact，1 = 主xact打开，2 = 
								 * 打开一个级别的子xact，等等 */
	bool		have_prep_stmt; /* 我们在这个xact中准备过任何语句吗？ */
	bool		have_error;		/* 在这个xact中，有任何子xacts中止吗？ */
	bool		changing_xact_state;	/* xact状态改变正在进行中 */
	bool		parallel_commit;	/* 我们是否并行提交（子）xacts？ */
	bool		invalidated;	/* 如果重新连接待定，则为真 */
	bool		keep_connections;	/* 设置keep_connections服务器选项的值 */
	Oid			serverid;		/* 用于获取服务器名称的外部服务器OID */
	uint32		server_hashvalue;	/* 外部服务器OID的哈希值 */
	uint32		mapping_hashvalue;	/* 用户映射OID的哈希值 */
	PgFdwConnState state;		/* 每连接的额外状态 */
} ConnCacheEntry;

/*
 * 连接缓存（在第一次使用时初始化）
 */
static HTAB *ConnectionHash = NULL;

/* 用于分配游标编号和准备语句编号 */
static unsigned int cursor_number = 0;
static unsigned int prep_stmt_number = 0;

/* 跟踪回调函数中是否需要任何工作 */
static bool xact_got_connection = false;

/*
 * SQL 函数
 */
PG_FUNCTION_INFO_V1(postgres_fdw_get_connections);
PG_FUNCTION_INFO_V1(postgres_fdw_disconnect);
PG_FUNCTION_INFO_V1(postgres_fdw_disconnect_all);

/* 私有函数的原型 */
static void fc_make_new_connection(ConnCacheEntry *fc_entry, UserMapping *fc_user);
static PGconn *fc_connect_pg_server(ForeignServer *fc_server, UserMapping *fc_user);
static void fc_disconnect_pg_server(ConnCacheEntry *fc_entry);
static void fc_check_conn_params(const char **fc_keywords, const char **fc_values, UserMapping *fc_user);
static void fc_configure_remote_session(PGconn *fc_conn);
static void fc_do_sql_command_begin(PGconn *fc_conn, const char *fc_sql);
static void fc_do_sql_command_end(PGconn *fc_conn, const char *fc_sql,
							   bool fc_consume_input);
static void fc_begin_remote_xact(ConnCacheEntry *fc_entry);
static void fc_pgfdw_xact_callback(XactEvent fc_event, void *fc_arg);
static void fc_pgfdw_subxact_callback(SubXactEvent fc_event,
								   SubTransactionId fc_mySubid,
								   SubTransactionId fc_parentSubid,
								   void *fc_arg);
static void fc_pgfdw_inval_callback(Datum fc_arg, int fc_cacheid, uint32 fc_hashvalue);
static void fc_pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *fc_entry);
static void fc_pgfdw_reset_xact_state(ConnCacheEntry *fc_entry, bool fc_toplevel);
static bool fc_pgfdw_cancel_query(PGconn *fc_conn);
static bool fc_pgfdw_exec_cleanup_query(PGconn *fc_conn, const char *fc_query,
									 bool fc_ignore_errors);
static bool fc_pgfdw_get_cleanup_result(PGconn *fc_conn, TimestampTz fc_endtime,
									 PGresult **fc_result, bool *fc_timed_out);
static void fc_pgfdw_abort_cleanup(ConnCacheEntry *fc_entry, bool fc_toplevel);
static void fc_pgfdw_finish_pre_commit_cleanup(List *fc_pending_entries);
static void fc_pgfdw_finish_pre_subcommit_cleanup(List *fc_pending_entries,
											   int fc_curlevel);
static bool fc_UserMappingPasswordRequired(UserMapping *fc_user);
static bool fc_disconnect_cached_connections(Oid fc_serverid);

/*
 * 获取一个PGconn，可以用于以用户授权的方式在远程PostgreSQL
 * 服务器上执行查询。如果我们没有合适的连接，将建立一个新连接，并在正确的子事务嵌套深度上打开事务，如果我们尚未这样做。
 *
 * 如果调用者打算创建任何准备语句，will_prep_stmt必须为真。由于这些在事务结束时不会自动消失（即使在错误情况下也是如此），我们需要这个标志来提示手动清理。
 *
 * 如果state不为NULL，*state接收与PGconn相关的每连接状态。
 */
PGconn * GetConnection(UserMapping *fc_user, bool fc_will_prep_stmt, PgFdwConnState **fc_state)
{
	bool		fc_found;
	bool		fc_retry = false;
	ConnCacheEntry *fc_entry;
	ConnCacheKey fc_key;
	MemoryContext fc_ccxt = CurrentMemoryContext;

	/* 第一次通过，初始化连接缓存哈希表 */
	if (ConnectionHash == NULL)
	{
		HASHCTL		fc_ctl;

		fc_ctl.keysize = sizeof(ConnCacheKey);
		fc_ctl.entrysize = sizeof(ConnCacheEntry);
		ConnectionHash = hash_create("postgres_fdw connections", 8,
									 &fc_ctl,
									 HASH_ELEM | HASH_BLOBS);

		/*
		 * 注册一些管理连接清理的回调函数。
		 * 这应该在每个后端中只做一次。
		 */
		RegisterXactCallback(fc_pgfdw_xact_callback, NULL);
		RegisterSubXactCallback(fc_pgfdw_subxact_callback, NULL);
		CacheRegisterSyscacheCallback(FOREIGNSERVEROID,
									  fc_pgfdw_inval_callback, (Datum) 0);
		CacheRegisterSyscacheCallback(USERMAPPINGOID,
									  fc_pgfdw_inval_callback, (Datum) 0);
	}

	/* 设置标志，表示我们在当前事务中进行了GetConnection */
	xact_got_connection = true;

	/* 为条目创建哈希键。假设密钥结构中没有填充字节 */
	fc_key = fc_user->umid;

	/*
	 * 查找或创建请求连接的缓存条目。
	 */
	fc_entry = hash_search(ConnectionHash, &fc_key, HASH_ENTER, &fc_found);
	if (!fc_found)
	{
		/*
		 * 我们只需要在这里清除“conn”；剩余字段将在“conn”设置时填充。
		 */
		fc_entry->conn = NULL;
	}

	/* 拒绝进一步使用失败的中止清理的连接。 */
	fc_pgfdw_reject_incomplete_xact_state_change(fc_entry);

	/*
	 * 如果由于无效而需要重新建立连接，则在退出所有事务后
	 * 尽快断开连接。
	 */
	if (fc_entry->conn != NULL && fc_entry->invalidated && fc_entry->xact_depth == 0)
	{
		elog(DEBUG3, "closing connection %p for option changes to take effect",
			 fc_entry->conn);
		fc_disconnect_pg_server(fc_entry);
	}

	/*
	 * 如果缓存条目没有连接，我们必须建立一个新的
	 * 连接。 （如果 connect_pg_server 抛出错误，缓存条目
	 * 将保持在有效空状态，即 conn == NULL。）
	 */
	if (fc_entry->conn == NULL)
		fc_make_new_connection(fc_entry, fc_user);

	/*
	 * 在使用缓存连接时，我们在这里检查其健康状况。
	 * 在我们退出所有事务的情况下，如果检测到连接损坏，
	 * 我们会尝试稍后重新建立新的连接。
	 */
	PG_TRY();
	{
		/* 处理任何待处理的异步请求。 */
		if (fc_entry->state.pendingAreq)
			process_pending_request(fc_entry->state.pendingAreq);
		/* 如有需要，开始新的事务或子事务。 */
		fc_begin_remote_xact(fc_entry);
	}
	PG_CATCH();
	{
		MemoryContext fc_ecxt = MemoryContextSwitchTo(fc_ccxt);
		ErrorData  *fc_errdata = CopyErrorData();

		/*
		 * 确定是否尝试重新建立连接。
		 *
		 * 在 libpq 检测到连接损坏后，任何其他错误
		 * 除了连接失败（例如，内存不足）都可能在 libpq
		 * 返回和预期的 ereport() 调用之间抛出
		 * pgfdw_report_error()。 在这种情况下，由于 PQstatus()
		 * 指示 CONNECTION_BAD，仅检查 PQstatus() 将导致
		 * 错误地检测到连接失败。 为了避免这种情况，我们还
		 * 验证错误的 sqlstate 是否为 ERRCODE_CONNECTION_FAILURE。
		 * 注意，仅检查 sqlstate 也可能导致另一个错误检测，
		 * 因为 pgfdw_report_error() 可能会报告任何 libpq
		 * 源自的错误条件的 ERRCODE_CONNECTION_FAILURE。
		 */
		if (fc_errdata->sqlerrcode != ERRCODE_CONNECTION_FAILURE ||
			PQstatus(fc_entry->conn) != CONNECTION_BAD ||
			fc_entry->xact_depth > 0)
		{
			MemoryContextSwitchTo(fc_ecxt);
			PG_RE_THROW();
		}

		/* 清理错误状态 */
		FlushErrorState();
		FreeErrorData(fc_errdata);
		fc_errdata = NULL;

		fc_retry = true;
	}
	PG_END_TRY();

	/*
	 * 如果检测到连接损坏，断开它，重新建立新的
	 * 连接并重试新的远程事务。如果再次报告连接失败，
	 * 我们放弃获取连接。
	 */
	if (fc_retry)
	{
		Assert(fc_entry->xact_depth == 0);

		ereport(DEBUG3,
				(errmsg_internal("could not start remote transaction on connection %p",
								 fc_entry->conn)),
				errdetail_internal("%s", pchomp(PQerrorMessage(fc_entry->conn))));

		elog(DEBUG3, "closing connection %p to reestablish a new one",
			 fc_entry->conn);
		fc_disconnect_pg_server(fc_entry);

		if (fc_entry->conn == NULL)
			fc_make_new_connection(fc_entry, fc_user);

		fc_begin_remote_xact(fc_entry);
	}

	/* 记住调用者是否将准备语句 */
	fc_entry->have_prep_stmt |= fc_will_prep_stmt;

	/* 如果调用者需要访问每个连接的状态，则返回它。 */
	if (fc_state)
		*fc_state = &fc_entry->state;

	return fc_entry->conn;
}

/*
 * 重置缓存连接条目中的所有瞬态状态字段，并
 * 建立到远程服务器的新连接。
 */
static void fc_make_new_connection(ConnCacheEntry *fc_entry, UserMapping *fc_user)
{
	ForeignServer *fc_server = GetForeignServer(fc_user->serverid);
	ListCell   *fc_lc;

	Assert(fc_entry->conn == NULL);

	/* 重置所有瞬态状态字段，以确保所有字段都是干净的 */
	fc_entry->xact_depth = 0;
	fc_entry->have_prep_stmt = false;
	fc_entry->have_error = false;
	fc_entry->changing_xact_state = false;
	fc_entry->invalidated = false;
	fc_entry->serverid = fc_server->serverid;
	fc_entry->server_hashvalue =
		GetSysCacheHashValue1(FOREIGNSERVEROID,
							  ObjectIdGetDatum(fc_server->serverid));
	fc_entry->mapping_hashvalue =
		GetSysCacheHashValue1(USERMAPPINGOID,
							  ObjectIdGetDatum(fc_user->umid));
	memset(&fc_entry->state, 0, sizeof(fc_entry->state));

	/*
	 * 确定是否保持我们即将建立的连接在事务结束后依然打开，
	 * 以便后续事务可以重用它。
	 *
	 * 默认情况下，所有对任何外部服务器的连接保持打开状态。
	 *
	 * 还要确定是否在（子）事务结束时并行提交在远程
	 * 服务器上打开的（子）事务，默认情况下这是禁用的。
	 *
	 * 注意：仅在建立新连接时确定这些设置就足够了，
	 * 因为如果这些设置被更改，连接将被关闭并
	 * 稍后重新建立。
	 */
	fc_entry->keep_connections = true;
	fc_entry->parallel_commit = false;
	foreach(fc_lc, fc_server->options)
	{
		DefElem    *fc_def = (DefElem *) lfirst(fc_lc);

		if (strcmp(fc_def->defname, "keep_connections") == 0)
			fc_entry->keep_connections = defGetBoolean(fc_def);
		else if (strcmp(fc_def->defname, "parallel_commit") == 0)
			fc_entry->parallel_commit = defGetBoolean(fc_def);
	}

	/* 现在尝试建立连接 */
	fc_entry->conn = fc_connect_pg_server(fc_server, fc_user);

	elog(DEBUG3, "new postgres_fdw connection %p for server \"%s\" (user mapping oid %u, userid %u)",
		 fc_entry->conn, fc_server->servername, fc_user->umid, fc_user->userid);
}

/*
 * 使用指定的服务器和用户映射属性连接到远程服务器。
 */
static PGconn * fc_connect_pg_server(ForeignServer *fc_server, UserMapping *fc_user)
{
	PGconn	   *volatile fc_conn = NULL;

	/*
	 * 使用 PG_TRY 块确保在错误时关闭连接。
	 */
	PG_TRY();
	{
		const char **fc_keywords;
		const char **fc_values;
		char	   *fc_appname = NULL;
		int			fc_n;

		/*
		 * 从ForeignServer和UserMapping的通用选项构造连接参数。
		 * （其中一些可能不是libpq选项，在这种情况下我们将浪费几个数组槽。）为
		 * application_name、fallback_application_name、client_encoding、结束标记添加4个额外槽。
		 */
		fc_n = list_length(fc_server->options) + list_length(fc_user->options) + 4;
		fc_keywords = (const char **) palloc(fc_n * sizeof(char *));
		fc_values = (const char **) palloc(fc_n * sizeof(char *));

		fc_n = 0;
		fc_n += ExtractConnectionOptions(fc_server->options,
									  fc_keywords + fc_n, fc_values + fc_n);
		fc_n += ExtractConnectionOptions(fc_user->options,
									  fc_keywords + fc_n, fc_values + fc_n);

		/*
		 * 如果设置了pgfdw_application_name，则将其作为application_name使用。
		 *
		 * PQconnectdbParams()按顺序处理参数数组。如果任何关键字重复，则使用最后的值。
		 * 因此，请注意，pgfdw_application_name必须在ForeignServer的选项之后添加到数组中，
		 * 以便覆盖在ForeignServer中设置的application_name。
		 */
		if (pgfdw_application_name && *pgfdw_application_name != '\0')
		{
			fc_keywords[fc_n] = "application_name";
			fc_values[fc_n] = pgfdw_application_name;
			fc_n++;
		}

		/*
		 * 搜索参数数组以查找application_name设置，如果找到，
		 * 则用状态信息替换其中的转义序列。数组是向后搜索的，因为
		 * 如果application_name被重复设置，则使用最后的值。
		 */
		for (int fc_i = fc_n - 1; fc_i >= 0; fc_i--)
		{
			if (strcmp(fc_keywords[fc_i], "application_name") == 0 &&
				*(fc_values[fc_i]) != '\0')
			{
				/*
				 * 如果这个application_name设置不是空字符串，
				 * 即使在替换任何转义序列后，也要使用它。
				 */
				fc_appname = process_pgfdw_appname(fc_values[fc_i]);
				if (fc_appname[0] != '\0')
				{
					fc_values[fc_i] = fc_appname;
					break;
				}

				/*
				 * 这个空的application_name没有被使用，所以我们将
				 * values[i]设置为NULL，然后继续搜索数组以找到
				 * 下一个。
				 */
				fc_values[fc_i] = NULL;
				pfree(fc_appname);
				fc_appname = NULL;
			}
		}

		/* 将"postgres_fdw"用作fallback_application_name */
		fc_keywords[fc_n] = "fallback_application_name";
		fc_values[fc_n] = "postgres_fdw";
		fc_n++;

		/* 设置client_encoding，以便libpq可以正确地转换编码。 */
		fc_keywords[fc_n] = "client_encoding";
		fc_values[fc_n] = GetDatabaseEncodingName();
		fc_n++;

		fc_keywords[fc_n] = fc_values[fc_n] = NULL;

		/* 验证连接参数集 */
		fc_check_conn_params(fc_keywords, fc_values, fc_user);

		/*
		 * 我们必须遵守 fd.c 对非虚拟文件描述符的限制。假设
		 * 一个 PGconn 代表一个长时间存在的 FD。 （在此处执行此操作还
		 * 确保在需要时关闭 VFD。）
		 */
		if (!AcquireExternalFD())
		{
#ifndef WIN32					/* 不能在 ereport() 宏中写 #if */
			ereport(ERROR,
					(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
					 errmsg("could not connect to server \"%s\"",
							fc_server->servername),
					 errdetail("There are too many open files on the local server."),
					 errhint("Raise the server's max_files_per_process and/or \"ulimit -n\" limits.")));
#else
			ereport(ERROR,
					(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
					 errmsg("could not connect to server \"%s\"",
							server->servername),
					 errdetail("There are too many open files on the local server."),
					 errhint("Raise the server's max_files_per_process setting.")));
#endif
		}

		/* 可以建立连接 */
		fc_conn = PQconnectdbParams(fc_keywords, fc_values, false);

		if (!fc_conn)
			ReleaseExternalFD();	/* 因为PG_CATCH块不会 */

		if (!fc_conn || PQstatus(fc_conn) != CONNECTION_OK)
			ereport(ERROR,
					(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
					 errmsg("could not connect to server \"%s\"",
							fc_server->servername),
					 errdetail_internal("%s", pchomp(PQerrorMessage(fc_conn)))));

		/*
		 * 检查非超级用户是否使用密码建立连接；
		 * 否则，他是在利用postgres服务器的用户身份。另请参见contrib/dblink中的
		 * dblink_security_check和check_conn_params。
		 */
		if (!superuser_arg(fc_user->userid) && fc_UserMappingPasswordRequired(fc_user) &&
			!PQconnectionUsedPassword(fc_conn))
			ereport(ERROR,
					(errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
					 errmsg("password is required"),
					 errdetail("Non-superuser cannot connect if the server does not request a password."),
					 errhint("Target server's authentication method must be changed or password_required=false set in the user mapping attributes.")));

		/* 为使用准备新的会话 */
		fc_configure_remote_session(fc_conn);

		if (fc_appname != NULL)
			pfree(fc_appname);
		pfree(fc_keywords);
		pfree(fc_values);
	}
	PG_CATCH();
	{
		/* 如果我们成功创建了PGconn数据结构，释放它 */
		if (fc_conn)
		{
			PQfinish(fc_conn);
			ReleaseExternalFD();
		}
		PG_RE_THROW();
	}
	PG_END_TRY();

	return fc_conn;
}

/*
 * 断开连接缓存条目的任何打开连接。
 */
static void fc_disconnect_pg_server(ConnCacheEntry *fc_entry)
{
	if (fc_entry->conn != NULL)
	{
		PQfinish(fc_entry->conn);
		fc_entry->conn = NULL;
		ReleaseExternalFD();
	}
}

/*
 * 如果password_required已定义并且对该用户映射为false，则返回true，
 * 否则返回false。该映射已被预先验证。
 */
static bool fc_UserMappingPasswordRequired(UserMapping *fc_user)
{
	ListCell   *fc_cell;

	foreach(fc_cell, fc_user->options)
	{
		DefElem    *fc_def = (DefElem *) lfirst(fc_cell);

		if (strcmp(fc_def->defname, "password_required") == 0)
			return defGetBoolean(fc_def);
	}

	return true;
}

/*
 * 对于非超级用户，坚决要求connstr指定密码。
 * 这防止密码从.pgpass、服务文件、环境等中提取。
 * 我们不希望postgres用户的密码、
 * 证书等对非超级用户可访问。（另请参见contrib/dblink中的
 * dblink_connstr_check。）
 */
static void fc_check_conn_params(const char **fc_keywords, const char **fc_values, UserMapping *fc_user)
{
	int			fc_i;

	/* 如果是超级用户则无需检查 */
	if (superuser_arg(fc_user->userid))
		return;

	/* 如果参数包含非空密码，则可以 */
	for (fc_i = 0; fc_keywords[fc_i] != NULL; fc_i++)
	{
		if (strcmp(fc_keywords[fc_i], "password") == 0 && fc_values[fc_i][0] != '\0')
			return;
	}

	/* 如果超级用户在用户映射创建时明确表示可以则可以 */
	if (!fc_UserMappingPasswordRequired(fc_user))
		return;

	ereport(ERROR,
			(errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
			 errmsg("password is required"),
			 errdetail("Non-superusers must provide a password in the user mapping.")));
}

/*
 * 发出 SET 命令以确保远程会话配置正确。
 *
 * 我们在连接时只做一次这件事，假设之后不会有任何改变
 * 值的操作。由于我们永远不会向远程发送临时函数调用，
 * 从我们这一端应该没有任何方法打破这个假设。
 * 可以想象一些方法打破它在远程端，例如让一个外部表指向一个
 * 包含 set_config 调用的视图 --- 但是一旦承认恶意视图定义的
 * 可能性，就有无数种方法可以破坏事情。
 */
static void fc_configure_remote_session(PGconn *fc_conn)
{
	int			fc_remoteversion = PQserverVersion(fc_conn);

	/* 强制搜索路径只包含 pg_catalog (参见 deparse.c) */
	do_sql_command(fc_conn, "SET search_path = pg_catalog");

	/*
	 * 设置远程时区；这基本上只是装饰性的，因为所有
	 * 传输和返回的 timestamptzs 应该明确指定一个时区
	 * 。然而，它使回归测试的输出更加可预测。
	 *
	 * 我们并不冒险将远程时区设置为与我们相同，因为远程
	 * 服务器可能使用不同的时区数据库。相反，使用 GMT
	 * （加引号，因为非常旧的服务器对大小写很挑剔）。这在
	 * 远程的时区数据库中无论如何都能保证有效，
	 * 因为 pg_tzset() 将其硬编码（至少在 PG 9.2 及以后的版本中）。
	 */
	do_sql_command(fc_conn, "SET timezone = 'GMT'");

	/*
	 * 设置值以确保从远程返回的数据输出明确无歧义。
	 * （此逻辑应与 pg_dump 做的相匹配。还请参见 postgres_fdw.c
	 * 中的 set_transmission_modes。）
	 */
	do_sql_command(fc_conn, "SET datestyle = ISO");
	if (fc_remoteversion >= 80400)
		do_sql_command(fc_conn, "SET intervalstyle = postgres");
	if (fc_remoteversion >= 90000)
		do_sql_command(fc_conn, "SET extra_float_digits = 3");
	else
		do_sql_command(fc_conn, "SET extra_float_digits = 2");
}

/*
 * 便利子例程，以向远程发出不返回数据的 SQL 命令
 */
void do_sql_command(PGconn *fc_conn, const char *fc_sql)
{
	fc_do_sql_command_begin(fc_conn, fc_sql);
	fc_do_sql_command_end(fc_conn, fc_sql, false);
}

static void fc_do_sql_command_begin(PGconn *fc_conn, const char *fc_sql)
{
	if (!PQsendQuery(fc_conn, fc_sql))
		pgfdw_report_error(ERROR, NULL, fc_conn, false, fc_sql);
}

static void fc_do_sql_command_end(PGconn *fc_conn, const char *fc_sql, bool fc_consume_input)
{
	PGresult   *fc_res;

	/*
	 * 如果请求，消耗从套接字获得的所有可用数据。
	 * （注意，如果所有数据都可用，这允许 pgfdw_get_result
	 * 调用 PQgetResult，而不强制 WaitLatchOrSocket 的开销，
	 * 这与 PQconsumeInput 的开销相比将是大的。）
	 */
	if (fc_consume_input && !PQconsumeInput(fc_conn))
		pgfdw_report_error(ERROR, NULL, fc_conn, false, fc_sql);
	fc_res = pgfdw_get_result(fc_conn, fc_sql);
	if (PQresultStatus(fc_res) != PGRES_COMMAND_OK)
		pgfdw_report_error(ERROR, fc_res, fc_conn, true, fc_sql);
	PQclear(fc_res);
}

/*
 * 启动远程事务或子事务（如果需要）。
 *
 * 请注意，我们在远程会话中始终使用至少 REPEATABLE READ。
 * 这样，如果查询发起多个对同一或不同外部表的扫描，
 * 我们将从这些扫描中获得快照一致的结果。一个缺点是
 * 我们无法提供合理的 READ COMMITTED 行为仿真 --- 如果
 * 我们有其他方法来控制哪些远程查询共享快照，那就好了。
 */
static void fc_begin_remote_xact(ConnCacheEntry *fc_entry)
{
	int			fc_curlevel = GetCurrentTransactionNestLevel();

	/* 如果我们还没有，启动主事务 */
	if (fc_entry->xact_depth <= 0)
	{
		const char *fc_sql;

		elog(DEBUG3, "starting remote transaction on connection %p",
			 fc_entry->conn);

		if (IsolationIsSerializable())
			fc_sql = "START TRANSACTION ISOLATION LEVEL SERIALIZABLE";
		else
			fc_sql = "START TRANSACTION ISOLATION LEVEL REPEATABLE READ";
		fc_entry->changing_xact_state = true;
		do_sql_command(fc_entry->conn, fc_sql);
		fc_entry->xact_depth = 1;
		fc_entry->changing_xact_state = false;
	}

	/*
	 * 如果我们处于子事务中，堆叠保存点以匹配我们的级别。
	 * 这确保我们可以在子事务中止时只回滚所需的效果。
	 */
	while (fc_entry->xact_depth < fc_curlevel)
	{
		char		fc_sql[64];

		snprintf(fc_sql, sizeof(fc_sql), "SAVEPOINT s%d", fc_entry->xact_depth + 1);
		fc_entry->changing_xact_state = true;
		do_sql_command(fc_entry->conn, fc_sql);
		fc_entry->xact_depth++;
		fc_entry->changing_xact_state = false;
	}
}

/*
 * 释放通过调用 GetConnection 创建的连接引用计数。
 */
void ReleaseConnection(PGconn *fc_conn)
{
	/*
	 * 目前，我们实际上并不跟踪连接引用，因为所有的
	 * 清理都是基于事务或子事务管理的。因此这里没有
	 * 任何要做的事情。
	 */
}

/*
 * 为游标分配一个“唯一”的号码。
 *
 * 这些号码只需要在事务内每个连接中是唯一的。
 * 目前我们忽略每个连接的这一点，而是在事务中的所有连接之间分配它们，
 * 但我们要求提供连接，以便我们可以在将来进行细化。
 *
 * 请注意，即使在非常长的事务中发生了环绕，实际的冲突也是高度不可能的；
 * 只需确保使用 %u 而不是 %d 来打印。
 */
unsigned int GetCursorNumber(PGconn *fc_conn)
{
	return ++cursor_number;
}

/*
 * 为预处理语句分配一个“唯一”的号码。
 *
 * 这与 GetCursorNumber 的工作方式非常相似，不同的是我们在会话中从不重置计数器。
 * 这是因为我们不能百分之百确定已经在所有连接上处理掉
 * 所有预处理语句，并且通过重置增加预处理语句名称冲突的风险并不值得。
 */
unsigned int GetPrepStmtNumber(PGconn *fc_conn)
{
	return ++prep_stmt_number;
}

/*
 * 提交一个查询并等待结果。
 *
 * 此函数可以被信号中断。
 *
 * 调用者负责对结果进行错误处理。
 */
PGresult * pgfdw_exec_query(PGconn *fc_conn, const char *fc_query, PgFdwConnState *fc_state)
{
	/* 首先，处理任何待处理的异步请求。 */
	if (fc_state && fc_state->pendingAreq)
		process_pending_request(fc_state->pendingAreq);

	/*
	 * 提交一个查询。由于我们不使用非阻塞模式，这也可能
	 * 阻塞。但其风险相对较小，因此我们暂时忽略这一点。
	 */
	if (!PQsendQuery(fc_conn, fc_query))
		pgfdw_report_error(ERROR, NULL, fc_conn, false, fc_query);

	/* 等待结果。 */
	return pgfdw_get_result(fc_conn, fc_query);
}

/*
 * 等待先前异步执行函数调用的结果。
 *
 * 此函数通过检查任何中断来提供快速响应。
 *
 * 此函数模拟 PQexec() 的行为，在有多个结果时返回最后一个结果。
 *
 * 调用者负责对结果进行错误处理。
 */
PGresult * pgfdw_get_result(PGconn *fc_conn, const char *fc_query)
{
	PGresult   *volatile fc_last_res = NULL;

	/* 接下来，不要在出现错误时泄漏任何 PGresults。 */
	PG_TRY();
	{
		for (;;)
		{
			PGresult   *fc_res;

			while (PQisBusy(fc_conn))
			{
				int			fc_wc;

				/* 睡眠，直到有事情可做 */
				fc_wc = WaitLatchOrSocket(MyLatch,
									   WL_LATCH_SET | WL_SOCKET_READABLE |
									   WL_EXIT_ON_PM_DEATH,
									   PQsocket(fc_conn),
									   -1L, PG_WAIT_EXTENSION);
				ResetLatch(MyLatch);

				CHECK_FOR_INTERRUPTS();

				/* 套接字中有数据可用吗？ */
				if (fc_wc & WL_SOCKET_READABLE)
				{
					if (!PQconsumeInput(fc_conn))
						pgfdw_report_error(ERROR, NULL, fc_conn, false, fc_query);
				}
			}

			fc_res = PQgetResult(fc_conn);
			if (fc_res == NULL)
				break;			/* 查询已完成 */

			PQclear(fc_last_res);
			fc_last_res = fc_res;
		}
	}
	PG_CATCH();
	{
		PQclear(fc_last_res);
		PG_RE_THROW();
	}
	PG_END_TRY();

	return fc_last_res;
}

/*
 * 报告我们从远程服务器获得的错误。
 *
 * elevel: 要使用的错误级别（通常为 ERROR，但可能更低）
 * res: 包含错误的 PGresult
 * conn: 我们执行查询的连接
 * clear: 如果为 true，则 PQclear 结果（否则调用者将处理它）
 * sql: NULL，或我们尝试执行的远程命令的文本
 *
 * 注意：选择不因远程错误抛出 ERROR 的调用者
 * 负责确保相关的 ConnCacheEntry 被标记为 have_error = true。
 */
void pgfdw_report_error(int fc_elevel, PGresult *fc_res, PGconn *fc_conn,
				   bool fc_clear, const char *fc_sql)
{
	/* 如果请求，必须在离开此函数之前释放 PGresult。 */
	PG_TRY();
	{
		char	   *fc_diag_sqlstate = PQresultErrorField(fc_res, PG_DIAG_SQLSTATE);
		char	   *fc_message_primary = PQresultErrorField(fc_res, PG_DIAG_MESSAGE_PRIMARY);
		char	   *fc_message_detail = PQresultErrorField(fc_res, PG_DIAG_MESSAGE_DETAIL);
		char	   *fc_message_hint = PQresultErrorField(fc_res, PG_DIAG_MESSAGE_HINT);
		char	   *fc_message_context = PQresultErrorField(fc_res, PG_DIAG_CONTEXT);
		int			fc_sqlstate;

		if (fc_diag_sqlstate)
			fc_sqlstate = MAKE_SQLSTATE(fc_diag_sqlstate[0],
									 fc_diag_sqlstate[1],
									 fc_diag_sqlstate[2],
									 fc_diag_sqlstate[3],
									 fc_diag_sqlstate[4]);
		else
			fc_sqlstate = ERRCODE_CONNECTION_FAILURE;

		/*
		 * 如果我们没有从 PGresult 收到消息，请尝试 PGconn。这
		 * 是必要的，因为对于连接级故障，PQexec 可能只返回 NULL，
		 * 而根本没有 PGresult。
		 */
		if (fc_message_primary == NULL)
			fc_message_primary = pchomp(PQerrorMessage(fc_conn));

		ereport(fc_elevel,
				(errcode(fc_sqlstate),
				 (fc_message_primary != NULL && fc_message_primary[0] != '\0') ?
				 errmsg_internal("%s", fc_message_primary) :
				 errmsg("could not obtain message string for remote error"),
				 fc_message_detail ? errdetail_internal("%s", fc_message_detail) : 0,
				 fc_message_hint ? errhint("%s", fc_message_hint) : 0,
				 fc_message_context ? errcontext("%s", fc_message_context) : 0,
				 fc_sql ? errcontext("remote SQL command: %s", fc_sql) : 0));
	}
	PG_FINALLY();
	{
		if (fc_clear)
			PQclear(fc_res);
	}
	PG_END_TRY();
}

/*
 * pgfdw_xact_callback --- 在主事务结束时进行清理。
 *
 * 这将运行得很晚，因此必须不进入本地的用户定义代码。
 * （在远程端进入此代码是可以的。其远程的
 * COMMIT TRANSACTION 可能会运行延迟触发器。）
 */
static void fc_pgfdw_xact_callback(XactEvent fc_event, void *fc_arg)
{
	HASH_SEQ_STATUS fc_scan;
	ConnCacheEntry *fc_entry;
	List	   *fc_pending_entries = NIL;

	/* 如果没有连接在此事务中被触及，则快速退出。 */
	if (!xact_got_connection)
		return;

	/*
	 * 扫描所有连接缓存条目以查找开放的远程事务，并
	 * 关闭它们。
	 */
	hash_seq_init(&fc_scan, ConnectionHash);
	while ((fc_entry = (ConnCacheEntry *) hash_seq_search(&fc_scan)))
	{
		PGresult   *fc_res;

		/* 如果现在没有开放连接，则忽略缓存条目 */
		if (fc_entry->conn == NULL)
			continue;

		/* 如果有开放的远程事务，则尝试关闭它 */
		if (fc_entry->xact_depth > 0)
		{
			elog(DEBUG3, "closing remote transaction on connection %p",
				 fc_entry->conn);

			switch (fc_event)
			{
				case XACT_EVENT_PARALLEL_PRE_COMMIT:
				case XACT_EVENT_PRE_COMMIT:

					/*
					 * 如果之前对此连接的中止清理失败，
					 * 我们无法对此进行更多命令。
					 */
					fc_pgfdw_reject_incomplete_xact_state_change(fc_entry);

					/* 在预提交期间提交所有远程事务 */
					fc_entry->changing_xact_state = true;
					if (fc_entry->parallel_commit)
					{
						fc_do_sql_command_begin(fc_entry->conn, "COMMIT TRANSACTION");
						fc_pending_entries = lappend(fc_pending_entries, fc_entry);
						continue;
					}
					do_sql_command(fc_entry->conn, "COMMIT TRANSACTION");
					fc_entry->changing_xact_state = false;

					/*
					 * 如果子事务中出现任何错误，并且我们
					 * 制作了预编译语句，则执行 DEALLOCATE ALL 以确保
					 * 我们摆脱所有预编译语句。这是
					 * 令人烦恼的，并且不是特别稳健，但不值得更努力地尝试。
					 *
					 * DEALLOCATE ALL 仅在 8.3 及更高版本中存在，因此这
					 * 限制了 postgres_fdw 可以与之通信的服务器的最大年龄。
					 * 我们故意忽略 DEALLOCATE 中的错误，以便我们能够与某些
					 * 较旧的服务器继续工作（在泄漏预编译语句的同时；但我们反正
					 * 不支持 8.3 之前的更新操作）。
					 */
					if (fc_entry->have_prep_stmt && fc_entry->have_error)
					{
						fc_res = PQexec(fc_entry->conn, "DEALLOCATE ALL");
						PQclear(fc_res);
					}
					fc_entry->have_prep_stmt = false;
					fc_entry->have_error = false;
					break;
				case XACT_EVENT_PRE_PREPARE:

					/*
					 * 我们不允许任何远程事务，因为在准备
					 * 事务提交之前保持这些事务开放并不合理。
					 * 目前，毫无条件地抛出错误；稍后我们可能会允许只读情况。
					 * 请注意，错误将导致我们立即返回
					 * 这里，事件 == XACT_EVENT_ABORT，因此我们将在那时清理
					 * 连接状态。
					 */
					ereport(ERROR,
							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
							 errmsg("cannot PREPARE a transaction that has operated on postgres_fdw foreign tables")));
					break;
				case XACT_EVENT_PARALLEL_COMMIT:
				case XACT_EVENT_COMMIT:
				case XACT_EVENT_PREPARE:
					/* 预提交应该已经关闭了开放的事务 */
					elog(ERROR, "missed cleaning up connection during pre-commit");
					break;
				case XACT_EVENT_PARALLEL_ABORT:
				case XACT_EVENT_ABORT:
					/* 在中止期间回滚所有远程事务 */
					fc_pgfdw_abort_cleanup(fc_entry, true);
					break;
			}
		}

		/* 重置状态以显示我们已经退出事务 */
		fc_pgfdw_reset_xact_state(fc_entry, true);
	}

	/* 如果有任何待处理连接，完成清理工作 */
	if (fc_pending_entries)
	{
		Assert(fc_event == XACT_EVENT_PARALLEL_PRE_COMMIT ||
			   fc_event == XACT_EVENT_PRE_COMMIT);
		fc_pgfdw_finish_pre_commit_cleanup(fc_pending_entries);
	}

	/*
	 * 无论事件类型如何，我们现在可以标记自己已退出
	 * 事务。（注意：如果我们在 PRE_COMMIT 或 PRE_PREPARE 期间来此，
	 * 这可以在 COMMIT 或 PREPARE 期间节省一遍多余的哈希表扫描。）
	 */
	xact_got_connection = false;

	/* 同时重置下一个事务的游标编号 */
	cursor_number = 0;
}

/*
 * pgfdw_subxact_callback --- 子事务结束时的清理工作。
 */
static void fc_pgfdw_subxact_callback(SubXactEvent fc_event, SubTransactionId fc_mySubid,
					   SubTransactionId fc_parentSubid, void *fc_arg)
{
	HASH_SEQ_STATUS fc_scan;
	ConnCacheEntry *fc_entry;
	int			fc_curlevel;
	List	   *fc_pending_entries = NIL;

	/* 在子事务开始或提交后没有任何操作。 */
	if (!(fc_event == SUBXACT_EVENT_PRE_COMMIT_SUB ||
		  fc_event == SUBXACT_EVENT_ABORT_SUB))
		return;

	/* 如果没有连接在此事务中被触及，则快速退出。 */
	if (!xact_got_connection)
		return;

	/*
	 * 扫描所有连接缓存条目以查找当前级别的开放远程子事务，
	 * 并关闭它们。
	 */
	fc_curlevel = GetCurrentTransactionNestLevel();
	hash_seq_init(&fc_scan, ConnectionHash);
	while ((fc_entry = (ConnCacheEntry *) hash_seq_search(&fc_scan)))
	{
		char		fc_sql[100];

		/*
		 * 我们只关心当前级别的开放远程子事务的连接。
		 */
		if (fc_entry->conn == NULL || fc_entry->xact_depth < fc_curlevel)
			continue;

		if (fc_entry->xact_depth > fc_curlevel)
			elog(ERROR, "missed cleaning up remote subtransaction at level %d",
				 fc_entry->xact_depth);

		if (fc_event == SUBXACT_EVENT_PRE_COMMIT_SUB)
		{
			/*
			 * 如果之前对此连接的中止清理失败，我们
			 * 无法对此进行更多命令。
			 */
			fc_pgfdw_reject_incomplete_xact_state_change(fc_entry);

			/* 在预提交期间提交所有远程子事务 */
			snprintf(fc_sql, sizeof(fc_sql), "RELEASE SAVEPOINT s%d", fc_curlevel);
			fc_entry->changing_xact_state = true;
			if (fc_entry->parallel_commit)
			{
				fc_do_sql_command_begin(fc_entry->conn, fc_sql);
				fc_pending_entries = lappend(fc_pending_entries, fc_entry);
				continue;
			}
			do_sql_command(fc_entry->conn, fc_sql);
			fc_entry->changing_xact_state = false;
		}
		else
		{
			/* 在中止期间回滚所有远程子事务 */
			fc_pgfdw_abort_cleanup(fc_entry, false);
		}

		/* 好的，我们已经退出该级别的子事务 */
		fc_pgfdw_reset_xact_state(fc_entry, false);
	}

	/* 如果有任何待处理连接，完成清理工作 */
	if (fc_pending_entries)
	{
		Assert(fc_event == SUBXACT_EVENT_PRE_COMMIT_SUB);
		fc_pgfdw_finish_pre_subcommit_cleanup(fc_pending_entries, fc_curlevel);
	}
}

/*
 * 连接失效回调函数
 *
 * 在对 pg_foreign_server 或 pg_user_mapping 目录条目进行更改后，
 * 如果当前事务尚未使用这些连接，请立即关闭依赖于该条目的连接。
 * 否则，将这些连接标记为无效，然后在当前事务结束时调用 pgfdw_xact_callback()
 * 关闭它们，因为在使用这些连接的事务中无法关闭它们。
 * 如果必要，关闭的连接将在下一个机会中重新创建。
 *
 * 尽管大多数缓存失效回调会无视给定的 hashvalue 而清除所有相关内容，
 * 但连接的开销足够大，因此值得尝试避免这种情况。
 *
 * 注意：我们可以通过检查单个选项值更严格地避免不必要的断开连接，
 * 但这似乎付出的努力与收益不成比例。
 */
static void fc_pgfdw_inval_callback(Datum fc_arg, int fc_cacheid, uint32 fc_hashvalue)
{
	HASH_SEQ_STATUS fc_scan;
	ConnCacheEntry *fc_entry;

	Assert(fc_cacheid == FOREIGNSERVEROID || fc_cacheid == USERMAPPINGOID);

	/* ConnectionHash 必须已经存在，如果我们已注册 */
	hash_seq_init(&fc_scan, ConnectionHash);
	while ((fc_entry = (ConnCacheEntry *) hash_seq_search(&fc_scan)))
	{
		/* 忽略无效条目 */
		if (fc_entry->conn == NULL)
			continue;

		/* hashvalue == 0 意味着缓存重置，必须清除所有状态 */
		if (fc_hashvalue == 0 ||
			(fc_cacheid == FOREIGNSERVEROID &&
			 fc_entry->server_hashvalue == fc_hashvalue) ||
			(fc_cacheid == USERMAPPINGOID &&
			 fc_entry->mapping_hashvalue == fc_hashvalue))
		{
			/*
			 * 如果该连接在此事务中尚未使用，则立即关闭连接。
			 * 否则将其标记为无效，以便 pgfdw_xact_callback() 可以在此
			 * 事务结束时关闭它。
			 */
			if (fc_entry->xact_depth == 0)
			{
				elog(DEBUG3, "discarding connection %p", fc_entry->conn);
				fc_disconnect_pg_server(fc_entry);
			}
			else
				fc_entry->invalidated = true;
		}
	}
}

/*
 * 如果给定的连接缓存条目被标记为处于 xact 状态更改的中间，
 * 则引发错误。此调用应发生在不期望有此类更改进行的时刻；
 * 如果发现有进展，表示我们在先前状态更改的中间中止了，
 * 现在不知道远程事务的实际状态。
 * 这样的连接无法安全地进一步使用。重新建立连接将改变快照
 * 并回滚已执行的任何写入，因此这也不是选择。因此，我们必须中止。
 */
static void fc_pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *fc_entry)
{
	ForeignServer *fc_server;

	/* 对于非活动条目和正常状态的条目无事可做 */
	if (fc_entry->conn == NULL || !fc_entry->changing_xact_state)
		return;

	/* 确保该条目处于非活动状态 */
	fc_disconnect_pg_server(fc_entry);

	/* 查找要在下面的消息中显示的服务器名称 */
	fc_server = GetForeignServer(fc_entry->serverid);

	ereport(ERROR,
			(errcode(ERRCODE_CONNECTION_EXCEPTION),
			 errmsg("connection to server \"%s\" was lost",
					fc_server->servername)));
}

/*
 * 重置状态以表明我们已脱离（子）事务。
 */
static void fc_pgfdw_reset_xact_state(ConnCacheEntry *fc_entry, bool fc_toplevel)
{
	if (fc_toplevel)
	{
		/* 重置状态以显示我们已经退出事务 */
		fc_entry->xact_depth = 0;

		/*
		 * 如果连接不处于良好的空闲状态，标记为无效或其服务器的
		 * keep_connections 选项被禁用，则丢弃它以进行恢复。
		 * 下一个 GetConnection 将打开一个新的连接。
		 */
		if (PQstatus(fc_entry->conn) != CONNECTION_OK ||
			PQtransactionStatus(fc_entry->conn) != PQTRANS_IDLE ||
			fc_entry->changing_xact_state ||
			fc_entry->invalidated ||
			!fc_entry->keep_connections)
		{
			elog(DEBUG3, "discarding connection %p", fc_entry->conn);
			fc_disconnect_pg_server(fc_entry);
		}
	}
	else
	{
		/* 重置状态以表明我们已脱离子事务 */
		fc_entry->xact_depth--;
	}
}

/*
 * 取消当前正在进行的查询（我们没有查询文本）并忽略结果。如果我们成功取消查询并丢弃任何待处理结果，则返回 true，否则返回 false。
 *
 * 如果我们在这里抛出一个 ERROR 也不是一个大问题，但如果我们进入错误递归麻烦，我们最终会关闭连接，这将导致整个顶层事务失败，即使使用了子事务。尽量在可以的情况下使用 WARNING。
 *
 * XXX：如果查询是由 fetch_more_data_begin() 发送的，我们可以从每个连接状态中保存的 pendingAreq 获取查询文本，然后使用它报告查询。
 */
static bool fc_pgfdw_cancel_query(PGconn *fc_conn)
{
	PGcancel   *fc_cancel;
	char		fc_errbuf[256];
	PGresult   *fc_result = NULL;
	TimestampTz fc_endtime;
	bool		fc_timed_out;

	/*
	 * 如果取消查询和丢弃结果花费太长时间，假设连接已断开。
	 */
	fc_endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);

	/*
	 * 发出取消请求。不幸的是，没有好的方法来限制我们可能在 PQgetCancel() 中阻塞的时间。
	 */
	if ((fc_cancel = PQgetCancel(fc_conn)))
	{
		if (!PQcancel(fc_cancel, fc_errbuf, sizeof(fc_errbuf)))
		{
			ereport(WARNING,
					(errcode(ERRCODE_CONNECTION_FAILURE),
					 errmsg("could not send cancel request: %s",
							fc_errbuf)));
			PQfreeCancel(fc_cancel);
			return false;
		}
		PQfreeCancel(fc_cancel);
	}

	/* 获取并丢弃查询的结果。 */
	if (fc_pgfdw_get_cleanup_result(fc_conn, fc_endtime, &fc_result, &fc_timed_out))
	{
		if (fc_timed_out)
			ereport(WARNING,
					(errmsg("could not get result of cancel request due to timeout")));
		else
			ereport(WARNING,
					(errcode(ERRCODE_CONNECTION_FAILURE),
					 errmsg("could not get result of cancel request: %s",
							pchomp(PQerrorMessage(fc_conn)))));

		return false;
	}
	PQclear(fc_result);

	return true;
}

/*
 * 在（子）中止清理期间提交查询，并等待最多 30 秒以获取结果。如果查询在没有错误的情况下执行，则返回值为 true。
 * 如果查询成功执行但返回错误，只有在 ignore_errors 设置的情况下返回值才为 true。如果查询无法发送或超时，则返回值为 false。
 *
 * 如果我们在这里抛出一个 ERROR 也不是一个大问题，但如果我们进入错误递归麻烦，我们最终会关闭连接，这将导致整个顶层事务失败，即使使用了子事务。尽量在可以的情况下使用 WARNING。
 */
static bool fc_pgfdw_exec_cleanup_query(PGconn *fc_conn, const char *fc_query, bool fc_ignore_errors)
{
	PGresult   *fc_result = NULL;
	TimestampTz fc_endtime;
	bool		fc_timed_out;

	/*
	 * 如果执行清理查询花费太长时间，假设连接已断开。这很可能就是我们最初中止的原因（例如，语句超时，用户取消），因此超时时间不应太长。
	 */
	fc_endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);

	/*
	 * 提交一个查询。由于我们不使用非阻塞模式，这也可能
	 * 阻塞。但其风险相对较小，因此我们暂时忽略这一点。
	 */
	if (!PQsendQuery(fc_conn, fc_query))
	{
		pgfdw_report_error(WARNING, NULL, fc_conn, false, fc_query);
		return false;
	}

	/* 获取查询的结果。 */
	if (fc_pgfdw_get_cleanup_result(fc_conn, fc_endtime, &fc_result, &fc_timed_out))
	{
		if (fc_timed_out)
			ereport(WARNING,
					(errmsg("could not get query result due to timeout"),
					 fc_query ? errcontext("remote SQL command: %s", fc_query) : 0));
		else
			pgfdw_report_error(WARNING, NULL, fc_conn, false, fc_query);

		return false;
	}

	/* 如果不成功，则发出警告。 */
	if (PQresultStatus(fc_result) != PGRES_COMMAND_OK)
	{
		pgfdw_report_error(WARNING, fc_result, fc_conn, true, fc_query);
		return fc_ignore_errors;
	}
	PQclear(fc_result);

	return true;
}

/*
 * 在中止清理期间获取正在进行的查询的结果。这可能是一个因事务中止而被中断的查询，也可能是作为事务中止的一部分发起的查询，以使远程端回到适当状态。
 *
 * endtime 是我们应该放弃并假设远程端已断开的时间。如果超时到期或发生连接故障，则返回 true，否则返回 false。设置 *result 除非发生超时。仅在超时到期时将 timed_out 设置为 true。
 */
static bool fc_pgfdw_get_cleanup_result(PGconn *fc_conn, TimestampTz fc_endtime, PGresult **fc_result,
						 bool *fc_timed_out)
{
	volatile bool fc_failed = false;
	PGresult   *volatile fc_last_res = NULL;

	*fc_timed_out = false;

	/* 接下来，不要在出现错误时泄漏任何 PGresults。 */
	PG_TRY();
	{
		for (;;)
		{
			PGresult   *fc_res;

			while (PQisBusy(fc_conn))
			{
				int			fc_wc;
				TimestampTz fc_now = GetCurrentTimestamp();
				long		fc_cur_timeout;

				/* 如果超时已过，放弃，否则获取休眠时间。 */
				fc_cur_timeout = TimestampDifferenceMilliseconds(fc_now, fc_endtime);
				if (fc_cur_timeout <= 0)
				{
					*fc_timed_out = true;
					fc_failed = true;
					goto exit;
				}

				/* 睡眠，直到有事情可做 */
				fc_wc = WaitLatchOrSocket(MyLatch,
									   WL_LATCH_SET | WL_SOCKET_READABLE |
									   WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
									   PQsocket(fc_conn),
									   fc_cur_timeout, PG_WAIT_EXTENSION);
				ResetLatch(MyLatch);

				CHECK_FOR_INTERRUPTS();

				/* 套接字中有数据可用吗？ */
				if (fc_wc & WL_SOCKET_READABLE)
				{
					if (!PQconsumeInput(fc_conn))
					{
						/* 连接故障 */
						fc_failed = true;
						goto exit;
					}
				}
			}

			fc_res = PQgetResult(fc_conn);
			if (fc_res == NULL)
				break;			/* 查询已完成 */

			PQclear(fc_last_res);
			fc_last_res = fc_res;
		}
exit:	;
	}
	PG_CATCH();
	{
		PQclear(fc_last_res);
		PG_RE_THROW();
	}
	PG_END_TRY();

	if (fc_failed)
		PQclear(fc_last_res);
	else
		*fc_result = fc_last_res;
	return fc_failed;
}

/*
 * 中止远程事务或子事务。
 *
 * 如果主事务被回滚，则“toplevel”应设置为true，否则为false。
 *
 * 成功时将entry->changing_xact_state设置为false，失败时设置为true。
 */
static void fc_pgfdw_abort_cleanup(ConnCacheEntry *fc_entry, bool fc_toplevel)
{
	char		fc_sql[100];

	/*
	 * 如果我们已经处于错误递归问题中，则不要尝试清理连接。
	 */
	if (in_error_recursion_trouble())
		fc_entry->changing_xact_state = true;

	/*
	 * 如果连接已经无法恢复，则不要再触碰它。
	 */
	if (fc_entry->changing_xact_state)
		return;

	/*
	 * 将此连接标记为正在更改事务状态的过程。
	 */
	fc_entry->changing_xact_state = true;

	/* 假设我们可能丢失了对已准备语句的跟踪 */
	fc_entry->have_error = true;

	/*
	 * 如果已经通过异步执行函数向远程服务器提交了命令，则
	 * 该命令可能尚未完成。检查命令是否仍在远程服务器上处理，
	 * 如果是，则请求取消该命令。
	 */
	if (PQtransactionStatus(fc_entry->conn) == PQTRANS_ACTIVE &&
		!fc_pgfdw_cancel_query(fc_entry->conn))
		return;					/* 无法取消正在运行的查询 */

	if (fc_toplevel)
		snprintf(fc_sql, sizeof(fc_sql), "ABORT TRANSACTION");
	else
		snprintf(fc_sql, sizeof(fc_sql),
				 "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d",
				 fc_entry->xact_depth, fc_entry->xact_depth);
	if (!fc_pgfdw_exec_cleanup_query(fc_entry->conn, fc_sql, false))
		return;					/* 无法中止远程（子）事务 */

	if (fc_toplevel)
	{
		if (fc_entry->have_prep_stmt && fc_entry->have_error &&
			!fc_pgfdw_exec_cleanup_query(fc_entry->conn,
									  "DEALLOCATE ALL",
									  true))
			return;				/* 清理已准备语句时出现问题 */

		fc_entry->have_prep_stmt = false;
		fc_entry->have_error = false;
	}

	/*
	 * 如果每个连接状态的pendingAreq不为NULL，则意味着
	 * fetch_more_data_begin()启动的异步获取没有成功完成，
	 * 因此在fetch_more_data()中未重置每个连接状态；
	 * 在这种情况下，在这里重置每个连接状态。
	 */
	if (fc_entry->state.pendingAreq)
		memset(&fc_entry->state, 0, sizeof(fc_entry->state));

	/* 如果一切正常，则解除changing_xact_state */
	fc_entry->changing_xact_state = false;
}

/*
 * 完成对每个已向远程服务器发送了COMMIT命令的连接的预提交清理。
 */
static void fc_pgfdw_finish_pre_commit_cleanup(List *fc_pending_entries)
{
	ConnCacheEntry *fc_entry;
	List	   *fc_pending_deallocs = NIL;
	ListCell   *fc_lc;

	Assert(fc_pending_entries);

	/*
	 * 获取每个待处理条目的COMMIT命令的结果
	 */
	foreach(fc_lc, fc_pending_entries)
	{
		fc_entry = (ConnCacheEntry *) lfirst(fc_lc);

		Assert(fc_entry->changing_xact_state);

		/*
		 * 我们可能已经在套接字上接收到结果，因此传递
		 * consume_input=true尝试先消费它
		 */
		fc_do_sql_command_end(fc_entry->conn, "COMMIT TRANSACTION", true);
		fc_entry->changing_xact_state = false;

		/* 如有必要，平行执行 DEALLOCATE ALL */
		if (fc_entry->have_prep_stmt && fc_entry->have_error)
		{
			/* 忽略错误（请参阅pgfdw_xact_callback中的说明） */
			if (PQsendQuery(fc_entry->conn, "DEALLOCATE ALL"))
			{
				fc_pending_deallocs = lappend(fc_pending_deallocs, fc_entry);
				continue;
			}
		}
		fc_entry->have_prep_stmt = false;
		fc_entry->have_error = false;

		fc_pgfdw_reset_xact_state(fc_entry, true);
	}

	/* 如果没有待处理条目，则不再进行进一步工作 */
	if (!fc_pending_deallocs)
		return;

	/*
	 * 获取每个待处理条目的DEALLOCATE命令的结果
	 */
	foreach(fc_lc, fc_pending_deallocs)
	{
		PGresult   *fc_res;

		fc_entry = (ConnCacheEntry *) lfirst(fc_lc);

		/* 忽略错误（请参阅pgfdw_xact_callback中的说明） */
		while ((fc_res = PQgetResult(fc_entry->conn)) != NULL)
		{
			PQclear(fc_res);
			/* 如果连接丢失则停止（否则将无限循环） */
			if (PQstatus(fc_entry->conn) == CONNECTION_BAD)
				break;
		}
		fc_entry->have_prep_stmt = false;
		fc_entry->have_error = false;

		fc_pgfdw_reset_xact_state(fc_entry, true);
	}
}

/*
 * 完成对每个已向远程服务器发送了RELEASE命令的连接的预提交清理。
 */
static void fc_pgfdw_finish_pre_subcommit_cleanup(List *fc_pending_entries, int fc_curlevel)
{
	ConnCacheEntry *fc_entry;
	char		fc_sql[100];
	ListCell   *fc_lc;

	Assert(fc_pending_entries);

	/*
	 * 获取每个待处理条目的RELEASE命令的结果
	 */
	snprintf(fc_sql, sizeof(fc_sql), "RELEASE SAVEPOINT s%d", fc_curlevel);
	foreach(fc_lc, fc_pending_entries)
	{
		fc_entry = (ConnCacheEntry *) lfirst(fc_lc);

		Assert(fc_entry->changing_xact_state);

		/*
		 * 我们可能已经在套接字上接收到结果，因此传递
		 * consume_input=true尝试先消费它
		 */
		fc_do_sql_command_end(fc_entry->conn, fc_sql, true);
		fc_entry->changing_xact_state = false;

		fc_pgfdw_reset_xact_state(fc_entry, false);
	}
}

/*
 * 列出活跃的外部服务器连接。
 *
 * 该函数不接受任何输入参数，并返回由以下值组成的记录集：
 * - server_name - 活跃连接的服务器名称。如果外部服务器已被删除，但连接仍然活跃，则输出中的服务器名称将为 NULL。
 * - valid - true/false，表示连接是否有效。
 *   注意，连接可能在 pgfdw_inval_callback 中失效。
 *
 * 当没有缓存连接时，不会返回任何记录。
 */
Datum postgres_fdw_get_connections(PG_FUNCTION_ARGS)
{
#define POSTGRES_FDW_GET_CONNECTIONS_COLS	2
	ReturnSetInfo *fc_rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
	HASH_SEQ_STATUS fc_scan;
	ConnCacheEntry *fc_entry;

	InitMaterializedSRF(fcinfo, 0);

	/* 如果缓存不存在，我们不返回任何记录 */
	if (!ConnectionHash)
		PG_RETURN_VOID();

	hash_seq_init(&fc_scan, ConnectionHash);
	while ((fc_entry = (ConnCacheEntry *) hash_seq_search(&fc_scan)))
	{
		ForeignServer *fc_server;
		Datum		fc_values[POSTGRES_FDW_GET_CONNECTIONS_COLS];
		bool		fc_nulls[POSTGRES_FDW_GET_CONNECTIONS_COLS];

		/* 我们只寻找开放的远程连接 */
		if (!fc_entry->conn)
			continue;

		fc_server = GetForeignServerExtended(fc_entry->serverid, FSV_MISSING_OK);

		MemSet(fc_values, 0, sizeof(fc_values));
		MemSet(fc_nulls, 0, sizeof(fc_nulls));

		/*
		 * 外部服务器可能在当前显式事务中已被删除。
		 * 当与之关联的连接在当前事务中使用时，无法在另一个会话中删除该服务器，如果尝试，则另一个会话中的删除查询会阻塞，直到当前事务结束。
		 *
		 * 即使服务器在当前事务中已被删除，缓存仍可能具有关联的活跃连接条目，可以将这些连接称为悬挂连接。
		 * 由于我们无法从系统目录中获取悬挂连接的服务器名称，因此在输出中显示服务器名称的 NULL 值。
		 *
		 * 我们本可以通过将服务器名称存储在缓存条目中而不是服务器 oid 来做到更好，这样它就可以在输出中使用。
		 * 但是，每个缓存条目中的服务器名称需要 64 字节的内存，这在有很多缓存连接时是巨大的，并且使用场景（即在显式当前事务中删除外部服务器）似乎很少见。因此，我们选择在输出中显示服务器名称的 NULL 值。
		 *
		 * 这样的悬挂连接要么在下次使用时关闭，要么在 pgfdw_xact_callback 中的当前显式事务结束时关闭。
		 */
		if (!fc_server)
		{
			/*
			 * 如果服务器在当前显式事务中被删除，则该条目将在删除服务器命令结束时在 pgfdw_inval_callback 中失效。
			 * 注意，该连接在 pgfdw_inval_callback 中不会关闭，因为它仍在当前显式事务中使用。因此，在这里进行断言。
			 */
			Assert(fc_entry->conn && fc_entry->xact_depth > 0 && fc_entry->invalidated);

			/* 如果未找到服务器名称，则显示 NULL */
			fc_nulls[0] = true;
		}
		else
			fc_values[0] = CStringGetTextDatum(fc_server->servername);

		fc_values[1] = BoolGetDatum(!fc_entry->invalidated);

		tuplestore_putvalues(fc_rsinfo->setResult, fc_rsinfo->setDesc, fc_values, fc_nulls);
	}

	PG_RETURN_VOID();
}

/*
 * 断开指定的缓存连接。
 *
 * 该函数会丢弃从本地会话到给定名称的外部服务器所建立的开放连接。注意，使用不同用户映射可以与给定服务器建立多个连接。
 * 如果在当前本地事务中使用这些连接，则不会断开它们，并会报告警告消息。该函数返回 true，如果至少断开一个连接，否则返回 false。如果未找到给定名称的外部服务器，则报告错误。
 */
Datum postgres_fdw_disconnect(PG_FUNCTION_ARGS)
{
	ForeignServer *fc_server;
	char	   *fc_servername;

	fc_servername = text_to_cstring(PG_GETARG_TEXT_PP(0));
	fc_server = GetForeignServerByName(fc_servername, false);

	PG_RETURN_BOOL(fc_disconnect_cached_connections(fc_server->serverid));
}

/*
 * 断开所有缓存的连接。
 *
 * 此函数放弃通过 postgres_fdw 从本地会话到外部服务器建立的所有打开连接。
 * 如果当前本地事务中使用了这些连接，则不会断开连接，并会报告警告信息。此函数
 * 如果至少断开一个连接则返回真，否则返回假。
 */
Datum postgres_fdw_disconnect_all(PG_FUNCTION_ARGS)
{
	PG_RETURN_BOOL(fc_disconnect_cached_connections(InvalidOid));
}

/*
 * 断开缓存连接的工具。
 *
 * 此函数扫描所有连接缓存条目并断开与指定的外部服务器 OID 匹配的
 * 打开的连接。如果指定了 InvalidOid，则断开所有缓存的连接。
 *
 * 对于每个在当前事务中使用且未关闭的连接，此函数会发出警告。 如果
 * 至少断开一个连接则返回真，否则返回假。
 *
 * 注意，此函数甚至会断开由同一本地会话中的其他用户使用不同
 * 用户映射建立的连接。这导致非超级用户也能够关闭
 * 同一本地会话中由超级用户建立的连接。
 *
 * XXX 目前我们不认为这样做存在任何安全风险。但我们应该
 * 对此设置一些限制，例如，防止非超级用户关闭由超级用户
 * 建立的连接，即使是在同一会话中？
 */
static bool fc_disconnect_cached_connections(Oid fc_serverid)
{
	HASH_SEQ_STATUS fc_scan;
	ConnCacheEntry *fc_entry;
	bool		fc_all = !OidIsValid(fc_serverid);
	bool		fc_result = false;

	/*
	 * 连接缓存哈希表在此会话中尚未初始化，因此返回假。
	 */
	if (!ConnectionHash)
		return false;

	hash_seq_init(&fc_scan, ConnectionHash);
	while ((fc_entry = (ConnCacheEntry *) hash_seq_search(&fc_scan)))
	{
		/* 如果现在没有打开的连接，则忽略缓存条目。 */
		if (!fc_entry->conn)
			continue;

		if (fc_all || fc_entry->serverid == fc_serverid)
		{
			/*
			 * 发出警告，因为要关闭的连接在
			 * 当前事务中使用，并且现在无法断开。
			 */
			if (fc_entry->xact_depth > 0)
			{
				ForeignServer *fc_server;

				fc_server = GetForeignServerExtended(fc_entry->serverid,
												  FSV_MISSING_OK);

				if (!fc_server)
				{
					/*
					 * 如果在当前事务中使用的连接在删除外部服务器时被删除，
					 * 则该连接必须在 DROP SERVER 命令结束时由
					 * pgfdw_inval_callback 标记为无效。
					 */
					Assert(fc_entry->invalidated);

					ereport(WARNING,
							(errmsg("cannot close dropped server connection because it is still in use")));
				}
				else
					ereport(WARNING,
							(errmsg("cannot close connection for server \"%s\" because it is still in use",
									fc_server->servername)));
			}
			else
			{
				elog(DEBUG3, "discarding connection %p", fc_entry->conn);
				fc_disconnect_pg_server(fc_entry);
				fc_result = true;
			}
		}
	}

	return fc_result;
}
